package study.rabbitmq.consumer.config;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.RabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * arguments键	值	意义
 * x-message-ttl	数字类型，标志时间，以豪秒为单位	标志队列中的消息存活时间，也就是说队列中的消息超过了制定时间会被删除
 * x-expires	数字类型，标志时间，以豪秒为单位	队列自身的空闲存活时间，当前的queue在指定的时间内，没有consumer、basic.get也就是未被访问，就会被删除。
 * x-max-length和x-max-length-bytes	数字	最大长度和最大占用空间，设置了最大长度的队列，在超过了最大长度后进行插入会删除之前插入的消息为本次的留出空间,相应的最大占用大小也是这个道理，当超过了这个大小的时候，会删除之前插入的消息为本次的留出空间。
 * x-dead-letter-exchange和x-dead-letter-routing-key	字符串	消息因为超时或超过限制在队列里消失，这样我们就丢失了一些消息，也许里面就有一些是我们做需要获知的。而rabbitmq的死信功能则为我们带来了解决方案。设置了dead letter exchange与dead letter routingkey（要么都设定，要么都不设定）那些因为超时或超出限制而被删除的消息会被推动到我们设置的exchange中，再根据routingkey推到queue中
 * x-max-priority	数字	队列所支持的优先级别，列如设置为5，表示队列支持0到5六个优先级别，5最高，0最低，当然这需要生产者在发送消息时指定消息的优先级别，消息按照优先级别从高到低的顺序分发给消费者
 * alternate-exchange		下面简称AE，当一个消息不能被route的时候，如果exchange设定了AE，则消息会被投递到AE。如果存在AE链，则会按此继续投递，直到消息被route或AE链结束或遇到已经尝试route过消息的AE。
 */
@Configuration
public class RabbitConfig {
    public static final String DIRECT_EXCHANGE = "DIRECT_EXCHANGE";
    public static final String TOPIC_EXCHANGE = "TOPIC_EXCHANGE";
    public static final String FANOUT_EXCHANGE = "FANOUT_EXCHANGE";
    public static final String DEAD_EXCHANGE = "DEAD_EXCHANGE";
    public static final String LIMIT_EXCHANGE = "LIMIT_EXCHANGE";

    public static final String PRIVATE_CHART_QUEUE = "PRIVATE_CHART_QUEUE";
    public static final String SUBCRIBE_QUEUE = "SUBCRIBE_QUEUE";
    public static final String ALL_QUEUE = "ALL_QUEUE";
    public static final String ALL_SENCOND_QUEUE = "ALL_SENCOND_QUEUE";
    public static final String DEAD_QUEUE = "DEAD_QUEUE";
    public static final String TTL_QUEUE = "TTL_QUEUE";
    public static final String LIMIT_QUEUE = "LIMIT_QUEUE";

    /**
     * 点对点直连交换机 rout key 需要完全匹配
     * @return
     */
    @Bean
    public DirectExchange directExchange() {
        return new DirectExchange(DIRECT_EXCHANGE);
    }

    /**
     * 死信交换机
     * 到死信交换机的条件：
     *              消息过期
     *              消费端拒收
     *              队列数量到达最大，前面的记录就会进入死信
     * @return
     */
    @Bean
    public DirectExchange deadExchange() {
        return new DirectExchange(DEAD_EXCHANGE);
    }

    /**
     * 主题交换机 rout key 可以通配符 * 表示以.为分隔的一个单词，#表示任意个.的单词  比如 ab.*可以匹配 ab.12,不能匹配ab.12.12  而ab.#可以匹配
     * @return
     */
    @Bean
    public TopicExchange topicExchange() {
        return new TopicExchange(TOPIC_EXCHANGE);
    }

    /**
     * 广播形式的交换机 不需要指定rout key
     * @return
     */
    @Bean
    public FanoutExchange fanoutExchange() {
        return new FanoutExchange(FANOUT_EXCHANGE);
    }

    /**
     * 限流交换机
     * @return
     */
    @Bean
    public DirectExchange limitExchange() {
        return new DirectExchange(LIMIT_EXCHANGE);
    }

    @Bean
    public Queue privateChart() {
        return new Queue(PRIVATE_CHART_QUEUE);
    }

    @Bean
    public Queue subcribe() {
        return new Queue(SUBCRIBE_QUEUE);
    }

    @Bean
    public Queue all() {
        return new Queue(ALL_QUEUE);
    }

    @Bean
    public Queue allSecond() {
        return new Queue(ALL_SENCOND_QUEUE);
    }


    /**
     * 消息限制专门接收队列
     * @return
     */
    @Bean
    public Queue limitQueue() {
        return new Queue(LIMIT_QUEUE);
    }

    /**
     * 延时队列
     * @return
     */
    @Bean
    public Queue ttlQueue() {
        return new Queue(TTL_QUEUE);
    }

    /**
     * 死信队列 作用就是当当前队列中有消息过期，可以根据route key 转发到其他绑定到该交换机队列
     * @return
     */
    @Bean
    public Queue deadQueue() {
        return QueueBuilder.durable(DEAD_QUEUE).withArgument("x-dead-letter-exchange",DEAD_EXCHANGE)
                .withArgument("x-dead-letter-routing-key","ttl").build();
    }

    @Bean
    public Binding directBingding(@Qualifier("directExchange") DirectExchange exchange,
                                  @Qualifier("privateChart") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("privateChart");
    }

    @Bean
    public Binding subcribeBingding(@Qualifier("topicExchange") TopicExchange exchange,
                                  @Qualifier("subcribe") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("subcribe.#");
    }

    @Bean
    public Binding allBingding(@Qualifier("fanoutExchange") FanoutExchange exchange,
                                  @Qualifier("all") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding allSendcondBingding(@Qualifier("fanoutExchange") FanoutExchange exchange,
                                  @Qualifier("allSecond") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange);
    }

    @Bean
    public Binding deadBinding(@Qualifier("deadExchange") DirectExchange exchange,
                               @Qualifier("deadQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("dead");
    }

    /**
     * 死信交换机转发队列
     * @param exchange
     * @param queue
     * @return
     */
    @Bean
    public Binding ttlBinding(@Qualifier("deadExchange") DirectExchange exchange,
                               @Qualifier("ttlQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("ttl");
    }

    /**
     * 限流队列和交换机绑定
     * @param exchange
     * @param queue
     * @return
     */
    @Bean
    public Binding limitBinding(@Qualifier("limitExchange") DirectExchange exchange,
                               @Qualifier("limitQueue") Queue queue) {
        return BindingBuilder.bind(queue).to(exchange).with("limit");
    }

    /**
     * 消费者 @RabbitListener(queues = RabbitConfig.LIMIT_QUEUE,containerFactory = "containerFactory")可以指定containerFactory
     * @param connectionFactory
     * @return
     */
    @Bean("containerFactory")
    public RabbitListenerContainerFactory containerFactory(ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory containerFactory = new SimpleRabbitListenerContainerFactory();
        containerFactory.setConnectionFactory(connectionFactory);
        // 设置为手动应答
        containerFactory.setAcknowledgeMode(AcknowledgeMode.MANUAL);
        // 限流 如果2个接收到了没返回ack就不接受后面的消息
        containerFactory.setPrefetchCount(2);
        return containerFactory;
    }


}
